package org.example.rxjava.create;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.internal.schedulers.IoScheduler;

import java.util.List;
import java.util.concurrent.TimeUnit;

//1 create operate
public class Main {

    /**
     * Create observers in a variety of ways
     * @param args lock
     */
    public static void main(String[] args) {
        create();
        empty();
        error();
        never();
        just();
        formArray();
        fromIterable();
        defer();
        timer();
        interval();
        range();
        while (true){}
    }

    //.1
    private static void create() {
        Observable<Object> objectObservable = Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
                emitter.onNext("Hello RxJava");
            }
        });

        Observer<Object> observer = new Observer<>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {
                System.out.println(o);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };

        objectObservable.subscribeOn(new IoScheduler())
                .observeOn(new IoScheduler());

        //订阅
        objectObservable.subscribe(observer);
    }

    //.2 only send onComplete
    private static void empty() {
        Observable.empty().subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("Hello Rxjava");
            }
        });
    }

    //.3 only send onError
    private static void error() {
        Observable.error(new Exception())
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("Hello Rxjava");
                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //.4 no send event
    private static void never() {
        Observable.never()
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //.5 sendALimitedNumberOfEvents
    private static void just() {
        Observable.just("Hello RxJava")
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println(s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //.6 from a list
    private static void formArray() {
        Observable.fromArray(new String[]{"Hello RxJava"})
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println(s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //.7 fromIterable
    private static void fromIterable() {
        Observable.fromIterable(List.of("Hello RxJava"))
                .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //2.1 defer TODO
    private static void defer(){
        Bean1 b = new Bean1();
        System.out.println(b);
        Observable<Bean1> just = Observable.just(b);

        Observable<Object> objectObservable = Observable.defer(() -> just);

        b = new Bean1();
        System.out.println(b);

        objectObservable.subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {
                System.out.println(o);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //2.2 timer
    private static void timer(){
        Observable<Long> timer = Observable.timer(0, TimeUnit.SECONDS);

        timer.subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                System.out.println(aLong);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //2.3 interval
    private static void interval(){
        //such as Java.util.Timer
        Observable<Long> interval = Observable.interval(1, 1, TimeUnit.SECONDS);
        interval.subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                System.out.println(aLong);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //2.4 range
    private static void range(){
        Observable<Integer> range = Observable.range(0, 5);
        range.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                System.out.println(integer);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    static class Bean1{

    }

}
